package com.hu.hbase.uti;

import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;

import java.io.IOException;
import java.util.Objects;

/**
 * @Author: hujianjun
 * @Date: 2021/2/26 0:23
 * @Describe:
 */
@Slf4j
public class HBaseClient {
    /**
     * 获取Admin对象实例
     *
     * @param zkHost：zookeeper连接地址
     * @return
     */
    public Admin getHBaseAdmin() {
        Configuration configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.quorum", "localhost");

        Admin admin = null;
        try {
            admin = ConnectionFactory.createConnection(configuration).getAdmin();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return admin;
    }

    /**
     * 关闭资源
     *
     * @param admin
     * @param connection
     */
    public void close(Admin admin, Connection connection) {
        if (Objects.nonNull(admin)) {
            try {
                admin.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        if (Objects.nonNull(connection)) {
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 判断指定表名是否存在
     *
     * @param tableName：表名
     * @return
     */
    public boolean isExistTable(Admin hBaseAdmin, String tableName) throws IOException {
        return hBaseAdmin.tableExists(TableName.valueOf(tableName));
    }

    /**
     * 创建指定表名
     *
     * @param tableName：表名
     * @return
     */
    public void createTable(Admin hBaseAdmin, String tableName, String... cfNames) throws IOException {
        if (Objects.isNull(cfNames) || cfNames.length == 0) {
            log.info("列族信息不存在！");
            return;
        }

        if (isExistTable(hBaseAdmin, tableName)) {
            log.info(tableName + "表已存在！");
            return;
        }

        HTableDescriptor tableDescriptor = new HTableDescriptor(TableName.valueOf(tableName));
        for (String cfName : cfNames) {
            HColumnDescriptor hColumnDescriptor = new HColumnDescriptor(cfName);
//            hColumnDescriptor.setMaxVersions(10);
            tableDescriptor.addFamily(hColumnDescriptor);
        }
        hBaseAdmin.createTable(tableDescriptor);
    }

    /**
     * 删除指定表名
     *
     * @param tableName：表名
     * @return
     */
    public void createTable(Admin hBaseAdmin, String tableName) throws IOException {
        if (!isExistTable(hBaseAdmin, tableName)) {
            log.info(tableName + "表不存在！");
            return;
        }
        TableName tn = TableName.valueOf(tableName);
        hBaseAdmin.disableTable(tn);
        hBaseAdmin.deleteTable(tn);
    }

    /**
     * 创建命名空间
     *
     * @param nameSpace：表名
     * @return
     *
     * 没有判断命名空间的方法，所以只能捕获
     */
    public void createNameSpace(Admin hBaseAdmin, String nameSpace) {
        NamespaceDescriptor namespaceDescriptor = NamespaceDescriptor.create(nameSpace).build();
        try {
            hBaseAdmin.createNamespace(namespaceDescriptor);
        } catch (NamespaceExistException e) {
            e.printStackTrace();
        }catch (IOException e) {
            e.printStackTrace();
        }
    }

}
